import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.udf

object Sparksql {
  def main(args: Array[String]): Unit = {
//    1. 创建SparkSession实例对象；

    val sparksql=SparkSession.builder()
      .master("local[2]")
      .config("spark.sql.shuffle.partitions","2")
      .appName("Sparksql")
      .getOrCreate()
    //    2. 加载jd-behavior.log日志数据，封装数据集DataFrame；
    val sql=sparksql.read
      .option("delimiter", ",")
      .schema("ip STRING,user_id INT,url STRING,datetime STRING")
      .csv("D:\\IdeaProjects\\zg6\\sparksql\\src\\main\\java\\jd-behavior.log")
    //    3. 数据及注册临时表：tbl_jd_logs，查询前10条数据，打印输出；
    sql.createTempView("tbl_jd_logs")
    sql.show(10)
    //    4. 分析1：京东每日的UV，降序排序，打印输出；
    val sql4=sparksql.sql(
      """
        |SELECT DATE_FORMAT(REPLACE(datetime,'/','-'),'yyyy-MM-dd') ts,url,count(DISTINCT ip) UV FROM tbl_jd_logs
        |GROUP BY DATE_FORMAT(REPLACE(datetime,'/','-'),'yyyy-MM-dd'),url
        |""".stripMargin)
    sql4.show()

//    5. 分析2：京东每日浏览量PV，降序排序，打印输出；
    val sql5=sparksql.sql(
      """
        |SELECT DATE_FORMAT(REPLACE(datetime,'/','-'),'yyyy-MM-dd') ts,url,count(1) PV FROM tbl_jd_logs
        |GROUP BY DATE_FORMAT(REPLACE(datetime,'/','-'),'yyyy-MM-dd'),url
        |""".stripMargin)
        sql5.show()
//    6. 分析3：京东每个url每日访客数UV，降序排序，打印输出；
    val sql6=sparksql.sql(
    """
      |SELECT DATE_FORMAT(REPLACE(datetime,'/','-'),'yyyy-MM-dd') ts
      |,url,count(DISTINCT ip) UV FROM tbl_jd_logs
      |GROUP BY DATE_FORMAT(REPLACE(datetime,'/','-'),'yyyy-MM-dd'),url
      |ORDER BY UV DESC
      |""".stripMargin)
      sql6.show()

//    7. 分析4：京东每个url每日浏览量PV，降序排序，保存MySQL表：result_pv；

      sql4.write
      .format("jdbc")
      .option("url", "jdbc:mysql://node101:3306")
      .option("dbtable", "db_canal.result_pv")
      .option("user", "root")
      .option("password", "123456")
      .save()
//    8. SparkSQL定义UDF函数：从字段datetime获取每天日期；

    val plusOne = udf((x: String) => x.replace('/','-').substring(0,10))
    sparksql.udf.register("plusOne", plusOne)
    //    9. 分析5：编写SQL语句，使用UDF函数，获取日期，打印输出；
    sparksql.sql("SELECT plusOne(datetime) FROM tbl_jd_logs").show()

    sparksql.stop()

  }
}
